"""
数据清洗处理器模块
该模块负责处理数据清洗的核心逻辑
"""
from typing import Optional
import pandas as pd
import re
import os
from datetime import datetime
from sqlalchemy.orm import Session
import logging
import json
import requests
import time
import numpy as np
import string
import unicodedata

# 尝试修复httpx代理问题
try:
    import httpx
    
    # 检查是否需要进行monkey patch
    original_client = httpx.Client
    test_client_args = {'proxies': None}
    try:
        # 尝试创建带有proxies参数的客户端
        test_client = original_client(**test_client_args)
        logging.info("httpx.Client支持proxies参数，无需修复")
    except TypeError:
        logging.warning("检测到httpx.Client不支持proxies参数，应用monkey patch")
        
        # 定义不接受proxies参数的Client类
        class ClientWithoutProxies(original_client):
            def __init__(self, **kwargs):
                if 'proxies' in kwargs:
                    logging.info("拦截并移除了proxies参数")
                    del kwargs['proxies']
                super().__init__(**kwargs)
        
        # 替换httpx.Client
        httpx.Client = ClientWithoutProxies
        
        # 同样处理AsyncClient
        original_async_client = httpx.AsyncClient
        
        class AsyncClientWithoutProxies(original_async_client):
            def __init__(self, **kwargs):
                if 'proxies' in kwargs:
                    logging.info("拦截并移除了异步客户端的proxies参数")
                    del kwargs['proxies']
                super().__init__(**kwargs)
        
        # 替换httpx.AsyncClient
        httpx.AsyncClient = AsyncClientWithoutProxies
        
        logging.info("成功应用httpx客户端monkey patch")
except Exception as e:
    logging.warning(f"应用httpx代理问题修复失败: {str(e)}")

from backend.data_clean.models import DataCleanTask, UnitConversion
from backend.config.models import Brand
from backend.data_clean.config_manager import ConfigManager, config_manager
from backend.data_clean.clean_product_data import generate_deepseek_prompt, prepare_product_data
from backend.data_clean.api_service import call_deepseek_api
from backend.data_clean.enhanced_brand_recognition import BrandRecognizer

# 创建日志记录器
logger = logging.getLogger(__name__)

def clean_product_data(df: pd.DataFrame, db: Session = None, use_llm: bool = True, llm_threshold: float = 0.9, force_llm: bool = False) -> pd.DataFrame:
    """
    清洗商品数据，提取结构化信息
    
    Args:
        df: 输入的DataFrame，需要包含'商品名称'列
        db: 数据库会话，用于初始化配置
        use_llm: 是否使用大模型进行数据清洗，默认为True
        llm_threshold: 使用大模型的阈值，默认为0.9表示置信度低于90%的数据会用大模型处理
        force_llm: 是否强制使用大模型，默认为False表示不强制所有数据都用大模型处理
        
    Returns:
        清洗后的DataFrame
    """
    start_time = time.time()
    logger.info(f"开始数据清洗，数据量: {len(df)}，use_llm={use_llm}，llm_threshold={llm_threshold}，force_llm={force_llm}")
    logger.info(f"数据清洗策略：{'使用混合方式，先规则后大模型' if use_llm else '仅使用规则方法'}")
    if use_llm and not force_llm:
        logger.info(f"将使用规则方法处理所有数据，对置信度低于{llm_threshold*100:.0f}%的数据使用大模型处理")
    elif use_llm and force_llm:
        logger.info("强制所有数据都使用大模型处理")
    
    # 强制重新加载配置管理器以确保使用最新数据
    if db is not None:
        logger.info("重新加载配置管理器以确保使用最新数据")
        config_manager.reload(db)
        logger.info(f"配置管理器已重新加载，当前品牌数量: {len(config_manager.brands)}")
    else:
        logger.warning("没有提供数据库会话，无法重新加载配置")
        # 如果未初始化则进行初始化
        if not hasattr(config_manager, '_is_initialized') or not config_manager._is_initialized:
            logger.info("尝试初始化配置管理器")
            config_manager.initialize()
    
    # 确保DataFrame包含必要的列
    if '商品名称' not in df.columns:
        logger.error("缺少'商品名称'列，无法进行数据清洗")
        raise ValueError("输入DataFrame必须包含'商品名称'列")
    
    # 处理价格列
    price_col = None
    possible_price_columns = ['商品价格', '价格', 'price', '单价']
    for col in possible_price_columns:
        if col in df.columns:
            price_col = col
            logger.info(f"找到价格列: {price_col}")
            # 确保价格列是数值类型
            df[price_col] = pd.to_numeric(df[price_col], errors='coerce')
            break
    
    if price_col is None:
        logger.warning("未找到价格列，将添加默认价格列")
        df['商品价格'] = 0.0
        price_col = '商品价格'
    
    # 记录配置信息
    logger.info(f"配置信息: 品牌数量={len(config_manager.brands)}, "
                f"单位数量={len(config_manager.units)}, "
                f"营销词数量={len(config_manager.marketing_words)}")
    
    # 记录原始DataFrame的列和形状
    logger.info(f"输入DataFrame - 列: {df.columns.tolist()}, 形状: {df.shape}")
    
    # 通过智能清洗处理数据
    logger.info("开始进行数据清洗，采用优化策略")
    logger.info(f"数据清洗参数: 使用大模型={use_llm}, 置信度阈值={llm_threshold:.2f} (低于此值用模型处理), 强制使用大模型={force_llm}")
    
    # 检测是否存在增强处理器
    try:
        from backend.data_clean.enhanced_processor import EnhancedProcessor
        enhanced_processor_available = True
        logger.info("检测到增强版数据处理器，将使用增强功能")
    except ImportError:
        enhanced_processor_available = False
        logger.info("未检测到增强版数据处理器，将使用标准处理方法")
    
    try:
        if enhanced_processor_available:
            # 使用增强版处理器
            processor = EnhancedProcessor(config_manager)
            
            if use_llm:
                # 使用混合处理
                logger.info("使用增强版混合处理方法")
                cleaned_df = processor.hybrid_process(df, use_llm=True, llm_threshold=llm_threshold, force_llm=force_llm)
            else:
                # 只使用规则处理
                logger.info("使用增强版规则处理方法")
                cleaned_df = processor.process_dataframe(df)
                cleaned_df['process_method'] = 'enhanced_rule'
        else:
            # 使用原有处理方法
            if use_llm:
                # 使用hybrid_data_cleaning进行处理
                logger.info(f"使用标准混合方式处理数据，置信度阈值: {llm_threshold}，强制使用大模型: {force_llm}")
                cleaned_df = hybrid_data_cleaning(df, db, use_model_threshold=llm_threshold, force_model=force_llm)
            else:
                # 如果不使用大模型，使用传统方法处理
                logger.info("使用标准规则方法处理")
                cleaned_df = ai_driven_data_cleaning(df, db)
                cleaned_df['process_method'] = 'rule'
    except Exception as e:
        logger.error(f"数据清洗过程中出错: {str(e)}")
        import traceback
        logger.error(f"异常堆栈: {traceback.format_exc()}")
        
        # 如果处理失败，尝试回退到传统处理方法
        logger.warning("尝试回退到传统处理方法")
        try:
            cleaned_df = ai_driven_data_cleaning(df, db)
            cleaned_df['process_method'] = 'fallback_rule'
        except Exception as e2:
            logger.error(f"传统处理方法也失败: {str(e2)}")
            # 如果所有方法都失败，返回原始DataFrame
            logger.error("所有处理方法都失败，返回原始DataFrame")
            cleaned_df = df.copy()
            # 添加必要的列
            cleaned_df['brand'] = ''
            cleaned_df['specification'] = ''
            cleaned_df['unit'] = ''
            cleaned_df['cleaned_name'] = cleaned_df['商品名称']
            cleaned_df['confidence_score'] = 0.0
            cleaned_df['process_method'] = 'failed'
    
    # 确保清洗后的DataFrame包含原始列
    for col in df.columns:
        if col not in cleaned_df.columns:
            cleaned_df[col] = df[col]
    
    # 标准化列名（确保与现有代码兼容）
    if 'spec_unit' in cleaned_df.columns and 'unit' not in cleaned_df.columns:
        cleaned_df['unit'] = cleaned_df['spec_unit']
    
    if 'brand_in_database' in cleaned_df.columns and 'brand_in_database' not in cleaned_df.columns:
        cleaned_df['brand_in_database'] = cleaned_df['brand_in_database']
    
    logger.info(f"数据清洗完成，耗时: {time.time() - start_time:.2f}秒，处理后数据量: {len(cleaned_df)}")
    
    return cleaned_df

def ai_driven_data_cleaning(df: pd.DataFrame, db: Session) -> pd.DataFrame:
    """
    使用AI驱动的两阶段处理流程进行商品数据清洗
    
    Args:
        df: 包含商品数据的DataFrame
        db: 数据库会话，用于加载配置
        
    Returns:
        处理后的DataFrame
    """
    # 创建新的DataFrame，保留原始数据
    result_df = df.copy(deep=True)
    logger.info(f"ai_driven_data_cleaning +++++++++++++++++++>>>>>>>>>>>>>>>>")
    # 确保"商品名称"列存在
    if '商品名称' not in result_df.columns:
        logger.error("数据中缺少'商品名称'列")
        raise ValueError("数据文件必须包含'商品名称'列")
    
    # 添加新列
    result_df['cleaned_name'] = ''      # 清洗后的商品名称
    result_df['brand'] = ''             # 品牌
    result_df['specification'] = ''     # 规格
    result_df['unit'] = ''              # 单位
    result_df['price'] = 0.0            # 价格
    result_df['standard_unit'] = ''     # 标准单位
    result_df['standard_price'] = 0.0   # 标准价格（每标准单位的价格）
    result_df['brand_in_database'] = False  # 品牌是否在数据库中
    result_df['confidence_score'] = 0.0 # 结果置信度
    
    # 第一阶段：数据分析阶段
    analysis_result = analyze_product_data(df, db)
    
    # 第二阶段：智能提取阶段
    result_df = extract_with_dynamic_rules(result_df, analysis_result)
    
    # 最终验证
    logger.info(f"清洗后的数据框信息: 形状={result_df.shape}, 列={result_df.columns.tolist()}")
    
    # 确保行数一致
    if len(result_df) != len(df):
        logger.error(f"清洗后的数据行数 ({len(result_df)}) 与原始数据行数 ({len(df)}) 不一致！")
    
    # 确保所有原始列都保留
    for col in df.columns:
        if col not in result_df.columns:
            logger.error(f"清洗后的数据缺少原始列: {col}")
    
    return result_df

def analyze_product_data(df: pd.DataFrame, db: Session) -> dict:
    """
    分析商品数据，识别模式和规律
    
    Args:
        df: 包含商品数据的DataFrame
        
    Returns:
        包含分析结果的字典
    """
    analysis_result = {
        'common_patterns': {},        # 常见模式
        'brand_patterns': [],         # 品牌模式
        'spec_patterns': [],          # 规格模式
        'marketing_terms': [],        # 营销词汇
        'category_markers': {},       # 类别标记
        'bracket_patterns': {},       # 括号模式
        'confidence_rules': [],       # 置信度规则
    }
    
    # 只分析"商品名称"列
    product_names = df['商品名称'].dropna().astype(str).tolist()
    if not product_names:
        return analysis_result
    
    # 1. 识别商品名称格式模式
    logger.info("分析商品名称格式模式...")
    
    # 1.1 分析前缀模式
    prefix_counter = {}
    for name in product_names:
        words = name.split()
        if len(words) >= 2:
            prefix = words[0]
            if prefix not in prefix_counter:
                prefix_counter[prefix] = 0
            prefix_counter[prefix] += 1
            
            # 检查两个词的前缀
            if len(words) >= 3:
                two_word_prefix = ' '.join(words[:2])
                if two_word_prefix not in prefix_counter:
                    prefix_counter[two_word_prefix] = 0
                prefix_counter[two_word_prefix] += 1
    
    # 筛选出频率较高的前缀（可能是品牌）
    threshold = max(3, len(product_names) * 0.05)  # 阈值：至少3个或5%的商品
    common_prefixes = {k: v for k, v in prefix_counter.items() if v >= threshold}
    analysis_result['common_patterns']['prefixes'] = common_prefixes
    
    # 识别可能的品牌模式
    for prefix, count in common_prefixes.items():
        if prefix in config_manager.brands:
            analysis_result['brand_patterns'].append({
                'pattern': f'^{re.escape(prefix)}\\b',
                'brand': prefix,
                'is_known': True,
                'frequency': count
            })
        elif count > threshold * 2:  # 如果出现频率特别高，也视为可能的品牌
            analysis_result['brand_patterns'].append({
                'pattern': f'^{re.escape(prefix)}\\b',
                'brand': prefix,
                'is_known': False,
                'frequency': count
            })
    
    # 1.2 分析括号标记
    bracket_patterns = [r'\[(.*?)\]', r'【(.*?)】', r'\((.*?)\)', r'（(.*?)）']
    bracket_contents = {}
    
    for pattern in bracket_patterns:
        bracket_contents[pattern] = []
        for name in product_names:
            matches = re.findall(pattern, name)
            bracket_contents[pattern].extend(matches)
    
    # 统计括号内容频率
    for pattern, contents in bracket_contents.items():
        content_counter = {}
        for content in contents:
            if content not in content_counter:
                content_counter[content] = 0
            content_counter[content] += 1
        
        # 只保留出现2次以上的内容
        filtered_contents = {k: v for k, v in content_counter.items() if v >= 2}
        if filtered_contents:
            analysis_result['bracket_patterns'][pattern] = filtered_contents
    
    # 1.3 分析规格模式
    spec_patterns = [
        r'约\s*(\d+\.?\d*)\s*(g|kg|ml|l|斤|千克|克|升|毫升|包|袋|个|条)(?:/(?:袋|包|盒))?',
        r'(\d+\.?\d*)\s*(g|kg|ml|l|斤|千克|克|升|毫升|包|袋|个|条)(?:/(?:袋|包|盒))?',
        r'(\d+\.?\d*)(g|kg|ml|l|斤|千克|克|升|毫升|包|袋|个|条)'
    ]
    
    for pattern in spec_patterns:
        matches_count = 0
        for name in product_names:
            if re.search(pattern, name):
                matches_count += 1
        
        # 如果模式匹配了足够多的商品名称
        if matches_count >= len(product_names) * 0.1:  # 至少10%的商品
            analysis_result['spec_patterns'].append({
                'pattern': pattern,
                'frequency': matches_count,
                'percentage': round(matches_count / len(product_names) * 100, 2)
            })
    
    # 1.4 识别营销词汇
    # 从配置中获取已知的营销词汇
    known_marketing_words = set(config_manager.marketing_words)
    
    # 尝试从数据中发现新的营销词汇
    word_freq = {}
    for name in product_names:
        words = re.findall(r'[\w\u4e00-\u9fa5]{2,}', name)  # 至少2个字符的词
        for word in words:
            if word not in word_freq:
                word_freq[word] = 0
            word_freq[word] += 1
    
    # 筛选可能的营销词（高频但不是商品本身）
    potential_marketing_words = []
    for word, freq in word_freq.items():
        # 如果一个词出现频率高但长度短，可能是营销词
        if freq >= len(product_names) * 0.08 and len(word) <= 4:
            # 检查该词是否可能是商品名称的核心部分
            is_core_product = False
            for name in product_names:
                # 如果该词占据了商品名称的大部分，则可能是商品本身而非营销词
                if word in name and len(word) > len(name) * 0.5:
                    is_core_product = True
                    break
            
            if not is_core_product and word not in known_marketing_words:
                potential_marketing_words.append({
                    'word': word,
                    'frequency': freq,
                    'percentage': round(freq / len(product_names) * 100, 2)
                })
    
    # 按频率排序并添加到结果中
    potential_marketing_words.sort(key=lambda x: x['frequency'], reverse=True)
    analysis_result['marketing_terms'] = [item['word'] for item in potential_marketing_words[:10]]
    
    # 1.5 分析商品类别标记
    # 例如[春菜]、[高原菜]等
    category_markers = {}
    for pattern, contents in analysis_result['bracket_patterns'].items():
        for content, freq in contents.items():
            if len(content) <= 5:  # 短文本可能是类别标记
                if content not in category_markers:
                    category_markers[content] = 0
                category_markers[content] += freq
    
    analysis_result['category_markers'] = {k: v for k, v in category_markers.items() if v >= 2}
    
    # 2. 生成置信度规则
    # 基于分析结果创建置信度评分规则
    analysis_result['confidence_rules'] = [
        {'rule': 'has_known_brand', 'points': 10, 'description': '商品包含已知品牌'},
        {'rule': 'has_clear_spec', 'points': 10, 'description': '商品包含明确规格'},
        {'rule': 'name_too_short', 'points': -20, 'description': '清洗后名称过短'},
        {'rule': 'too_many_marketing_words', 'points': -5, 'description': '包含过多营销词'},
        {'rule': 'standard_format', 'points': 15, 'description': '符合标准格式：品牌+品名+规格'}
    ]
    
    logger.info(f"数据分析完成，发现 {len(analysis_result['brand_patterns'])} 个品牌模式，"
                f"{len(analysis_result['spec_patterns'])} 个规格模式，"
                f"{len(analysis_result['marketing_terms'])} 个营销词，"
                f"{len(analysis_result['category_markers'])} 个类别标记")
    
    return analysis_result

def extract_with_dynamic_rules(df: pd.DataFrame, analysis_result: dict) -> pd.DataFrame:
    """
    根据分析结果使用动态规则提取商品信息
    
    Args:
        df: 包含商品数据的DataFrame
        analysis_result: 数据分析结果
        
    Returns:
        处理后的DataFrame
    """
    result_df = df.copy(deep=True)
    
    # 初始化BrandRecognizer（用config_manager.brands作为品牌库）
    brand_recognizer = BrandRecognizer(brands=config_manager.brands)
    
    # 获取常用配置和规则
    price_pattern = r'(?:￥|\¥|\$)?(\d+(?:\.\d+)?)\s*(?:元|块|圆|RMB)?'  # 匹配价格
    spec_pattern = config_manager.get_specification_pattern_with_units()
    clean_chars_pattern = config_manager.clean_chars_pattern
    unit_conversion = config_manager.unit_conversion
    standard_unit_map = config_manager.standard_unit_map
    
    # 从配置管理器获取所有单位并创建扩展规格模式
    units_list = config_manager.units
    # 动态创建扩展规格模式的正则表达式
    if units_list:
        units_regex = '|'.join(map(re.escape, units_list))
        extended_spec_pattern = fr'\d+\.?\d*\s*(?:{units_regex})(?:/(?:袋|包|盒))?'
        logger.info(f"动态生成的扩展规格模式: {extended_spec_pattern}")
    else:
        # 如果单位列表为空，使用默认模式
        extended_spec_pattern = r'\d+\.?\d*\s*(?:g|kg|ml|l|斤|千克|克|升|毫升|包|袋|个|条|根|瓶|盒)(?:/(?:袋|包|盒))?'
        logger.warning("单位列表为空，使用默认扩展规格模式")
    
    # 记录配置信息，用于调试
    logger.info(f"单位转换规则: {list(unit_conversion.keys())}")
    logger.info(f"标准单位映射: {standard_unit_map}")
    
    # 从分析结果中提取的动态规则
    dynamic_brand_patterns = analysis_result['brand_patterns']
    dynamic_spec_patterns = [item['pattern'] for item in analysis_result['spec_patterns']]
    dynamic_marketing_terms = set(config_manager.marketing_words + analysis_result['marketing_terms'])
    dynamic_category_markers = analysis_result['category_markers']
    confidence_rules = analysis_result['confidence_rules']
    
    # 添加提取方式列
    result_df['extract_method'] = 'rule'
    
    # 应用动态规则处理每一行数据
    for idx, row in result_df.iterrows():
        orig_name = str(row['商品名称']) if not pd.isna(row['商品名称']) else ''
        # 使用BrandRecognizer进行品牌提取
        brand_result = brand_recognizer.extract_and_clean_brands(orig_name)
        # 写入多品牌相关字段
        result_df.at[idx, 'brand'] = brand_result.get('brand_string', '')
        result_df.at[idx, 'brand_list'] = [b['brand'] for b in brand_result.get('brands', [])]
        result_df.at[idx, 'cleaned_name'] = brand_result.get('cleaned_text', orig_name)
        result_df.at[idx, 'brand_extract_method'] = 'BrandRecognizer'
        continue
        
        # 确保商品名称是字符串
        orig_name = str(row['商品名称']) if not pd.isna(row['商品名称']) else ''
        confidence = 100.0  # 初始置信度
        
        # 记录应用的规则
        applied_rules = []

        # 1. 预处理 - 移除类别标记
        processed_name = orig_name
        for category in dynamic_category_markers.keys():
            # 移除各种括号中的类别标记
            patterns = [
                f'\\[{re.escape(category)}\\]',
                f'【{re.escape(category)}】',
                f'\\({re.escape(category)}\\)',
                f'（{re.escape(category)}）'
            ]
            for pattern in patterns:
                if re.search(pattern, processed_name):
                    processed_name = re.sub(pattern, ' ', processed_name)
                    applied_rules.append(f"removed_category_marker_{category}")
        
        # 清理多余的空格和特殊字符
        name = re.sub(clean_chars_pattern, ' ', processed_name).strip()
        
        # 2. 品牌识别 - 优化策略，优先匹配已知品牌
        brand_found = False
        
        # 2.0 优先直接匹配数据库中的品牌（无需考虑空格分隔）
        if not brand_found:
            # 按品牌名称长度排序，优先匹配较长的品牌名，避免部分匹配问题
            sorted_brands = sorted(config_manager.brands, key=len, reverse=True)
            for brand in sorted_brands:
                if brand in name:
                    result_df.at[idx, 'brand'] = brand
                    result_df.at[idx, 'brand_in_database'] = True
                    # 品牌替换时使用空格分隔，便于后续处理
                    name = re.sub(r'\b' + re.escape(brand) + r'\b', ' ', name, 1).strip()
                    # 如果品牌前后没有空格分隔，也尝试替换
                    if brand in name:  
                        name = name.replace(brand, ' ', 1).strip()
                    brand_found = True
                    confidence += 10
                    applied_rules.append("found_brand_direct_match")
                    break
        
        # 2.1 复合品牌处理
        if not brand_found:
            compound_brand_pattern = r'([\w\u4e00-\u9fa5]+)[×xX]([\w\u4e00-\u9fa5]+)'
            compound_match = re.search(compound_brand_pattern, name)
            
            if compound_match:
                brand1 = compound_match.group(1)
                brand2 = compound_match.group(2)
                
                # 检查两个品牌是否在数据库中
                brand1_in_db = brand1 in config_manager.brands
                brand2_in_db = brand2 in config_manager.brands
                
                # 根据品牌是否在数据库中决定保留哪个品牌
                if brand1_in_db and brand2_in_db:
                    # 如果两个品牌都在数据库中，组合它们
                    combined_brand = f"{brand1} {brand2}"
                    result_df.at[idx, 'brand'] = combined_brand
                    result_df.at[idx, 'brand_in_database'] = True
                    # 从名称中移除完整的复合品牌表达式
                    full_match = compound_match.group(0)
                    name = name.replace(full_match, ' ', 1).strip()
                    brand_found = True
                    confidence += 15  # 找到两个已知品牌，增加更多置信度
                    applied_rules.append("found_dual_known_brands")
                elif brand1_in_db:
                    # 如果只有第一个品牌在数据库中
                    result_df.at[idx, 'brand'] = brand1
                    result_df.at[idx, 'brand_in_database'] = True
                    # 从名称中移除复合品牌表达式
                    full_match = compound_match.group(0)
                    name = name.replace(full_match, ' ', 1).strip()
                    brand_found = True
                    confidence += 10
                    applied_rules.append("found_compound_brand_first_known")
                elif brand2_in_db:
                    # 如果只有第二个品牌在数据库中
                    result_df.at[idx, 'brand'] = brand2
                    result_df.at[idx, 'brand_in_database'] = True
                    # 从名称中移除复合品牌表达式
                    full_match = compound_match.group(0)
                    name = name.replace(full_match, ' ', 1).strip()
                    brand_found = True
                    confidence += 10
                    applied_rules.append("found_compound_brand_second_known")
        
        # 2.2 应用动态品牌模式
        if not brand_found:
            for brand_info in dynamic_brand_patterns:
                pattern = brand_info['pattern']
                brand_name = brand_info['brand']
                is_known = brand_info['is_known']
                
                brand_match = re.search(pattern, name)
                if brand_match:
                    # 匹配到的文本可能与模式定义的品牌略有不同
                    matched_text = brand_match.group(0).strip()
                    result_df.at[idx, 'brand'] = brand_name
                    result_df.at[idx, 'brand_in_database'] = is_known
                    name = name.replace(matched_text, ' ', 1).strip()
                    brand_found = True
                    
                    # 根据品牌是否已知调整置信度
                    if is_known:
                        confidence += 10
                        applied_rules.append("found_known_brand")
                    else:
                        confidence += 5
                        applied_rules.append("found_potential_brand")
                    break
        
        # 3. 营销词去除
        marketing_words_found = []
        for word in dynamic_marketing_terms:
            if word in name:
                name = name.replace(word, ' ')
                marketing_words_found.append(word)
        
        if marketing_words_found:
            applied_rules.append(f"removed_marketing_words_{len(marketing_words_found)}")
            if len(marketing_words_found) > 2:
                confidence -= 5
                applied_rules.append("too_many_marketing_words")
        
        # 4. 规格提取和价格处理
        spec_match = False
        spec_value = 0.0
        spec_unit = ''
        spec_text = ''
        price = 0.0
        
        # 4.1 价格提取
        price = 0.0
        # 如果原始数据中已有价格列，直接使用该价格
        if '商品价格' in row and not pd.isna(row['商品价格']):
            try:
                price = float(row['商品价格'])
                result_df.at[idx, 'price'] = price
                applied_rules.append("used_original_price")
            except (ValueError, TypeError):
                logger.warning(f"无法解析原始价格列的值: {row['商品价格']}")
                confidence -= 5
                applied_rules.append("price_parse_failed")
        # 如果没有价格列或价格为空，尝试从名称中提取
        elif price_match := re.search(price_pattern, name):
            try:
                price = float(price_match.group(1))
                result_df.at[idx, 'price'] = price
                name = name.replace(price_match.group(0), ' ').strip()
                applied_rules.append("extracted_price")
            except ValueError:
                logger.warning(f"无法解析价格: {price_match.group(0)}")
                confidence -= 5
                applied_rules.append("price_parse_failed")
        
        # 4.2 规格提取 - 按优先级依次尝试不同的规格提取方法
        
        # 4.2.1 首先尝试使用动态规格模式
        for pattern in dynamic_spec_patterns:
            matches = re.findall(pattern, name)
            if matches:
                match = matches[-1]  # 使用最后一个匹配
                
                # 根据模式类型处理匹配结果
                if isinstance(match, tuple) and len(match) >= 2:
                    try:
                        spec_value = float(match[0])
                        spec_unit = match[1]
                        
                        # 构建完整的规格文本
                        full_match_text = re.search(pattern, name).group()
                        spec_text = full_match_text
                        result_df.at[idx, 'specification'] = full_match_text
                        result_df.at[idx, 'unit'] = spec_unit
                        
                        # 从名称中移除规格
                        name = name.replace(full_match_text, ' ').strip()
                        spec_match = True
                        confidence += 10
                        applied_rules.append("extracted_spec_with_dynamic_pattern")
                        
                        # 检查是否发现了新的单位
                        if spec_unit not in config_manager.units:
                            logger.info(f"发现新单位: {spec_unit}")
                            # 记录下新单位，便于后续添加到配置中
                            if not hasattr(config_manager, 'potential_new_units'):
                                config_manager.potential_new_units = {}
                            if spec_unit not in config_manager.potential_new_units:
                                config_manager.potential_new_units[spec_unit] = 0
                            config_manager.potential_new_units[spec_unit] += 1
                        
                        break
                    except (ValueError, IndexError):
                        continue
        
        # 4.2.2 如果动态模式未匹配，尝试使用扩展模式（针对简单规格如"2根"）
        if not spec_match:
            extended_match = re.search(extended_spec_pattern, name)
            if extended_match:
                spec_text = extended_match.group().strip()
                result_df.at[idx, 'specification'] = spec_text
                name = name.replace(spec_text, ' ').strip()
                
                # 提取规格中的数值和单位
                spec_num_match = re.search(r'\d+\.?\d*', spec_text)
                spec_unit_match = re.search(r'[^\d\s\.]+', spec_text)
                
                if spec_num_match and spec_unit_match:
                    try:
                        spec_value = float(spec_num_match.group())
                        spec_unit = spec_unit_match.group().strip()
                        result_df.at[idx, 'unit'] = spec_unit
                        
                        # 记录新单位
                        if spec_unit not in config_manager.units:
                            logger.info(f"发现新单位: {spec_unit}")
                            if not hasattr(config_manager, 'potential_new_units'):
                                config_manager.potential_new_units = {}
                            if spec_unit not in config_manager.potential_new_units:
                                config_manager.potential_new_units[spec_unit] = 0
                            config_manager.potential_new_units[spec_unit] += 1
                        
                        spec_match = True
                        applied_rules.append("extracted_spec_with_extended_pattern")
                        confidence += 8
                    except ValueError:
                        logger.warning(f"无法解析规格数值: {spec_text}")
                        confidence -= 5
                        applied_rules.append("spec_parse_failed")
        
        # 5. 标准化商品名称
        # 使用正则表达式清理名称中的特殊字符
        name = re.sub(r'[%@#&:;.,\*]+', ' ', name).strip()
        # 合并多个空格为一个
        name = re.sub(r'\s+', ' ', name).strip()
        
        # 6. 标准化单位和价格
        standard_unit = ''
        standard_price = 0.0
        standard_unit2 = ''
        standard_price2 = 0.0
        standard_value = 0.0
        
        if spec_value > 0 and spec_unit and spec_unit in unit_conversion:
            try:
                # 计算标准单位值
                standard_value = unit_conversion[spec_unit](spec_value)
                
                # 记录转换过程
                if idx < 5 or idx % 100 == 0:
                    logger.info(f"单位转换: {spec_value}{spec_unit} -> {standard_value}")
                
                # 设置标准单位
                if spec_unit in standard_unit_map:
                    standard_unit = standard_unit_map[spec_unit]
                    result_df.at[idx, 'standard_unit'] = standard_unit
                    applied_rules.append("standardized_unit")
                
                # 计算标准价格
                if price > 0 and standard_value > 0:
                    standard_price = price / standard_value
                    result_df.at[idx, 'standard_price'] = round(standard_price, 2)
                    
                    # 处理复合规格，例如"500ml*6瓶整箱"
                    if '*' in spec_text and re.search(r'\d+\s*[瓶盒袋个包]', spec_text):
                        # 尝试提取件数，例如6瓶
                        pieces_match = re.search(r'(\d+)\s*([瓶盒袋个包])', spec_text)
                        if pieces_match:
                            pieces_num = int(pieces_match.group(1))
                            pieces_unit = pieces_match.group(2)
                            
                            # 设置第二个标准单位和价格
                            standard_unit2 = pieces_unit
                            standard_price2 = price / pieces_num
                            result_df.at[idx, 'standard_unit2'] = standard_unit2
                            result_df.at[idx, 'standard_price2'] = round(standard_price2, 2)
                            applied_rules.append("calculated_secondary_unit_price")
                            
                            if idx < 5 or idx % 100 == 0:
                                logger.info(f"第二单位价格: 原价={price}, 每{pieces_unit}价格={standard_price2}")
                    
                    # 记录价格计算过程
                    if idx < 5 or idx % 100 == 0:
                        logger.info(f"价格转换: 原价={price}, 标准价格={standard_price}/每{standard_unit}")
            except Exception as e:
                # 详细记录错误
                logger.warning(f"单位转换或价格计算失败: {str(e)}")
                logger.warning(f"单位={spec_unit}, 原始规格值={spec_value}, 价格={price}")
                confidence -= 10
                applied_rules.append("unit_conversion_failed")
        
        # 7. 设置清洗后的商品名称
        # 防止名称为空
        if not name:
            # 如果前面的处理导致名称为空，尝试重新提取
            parts = orig_name.split()
            
            # 去除已提取的品牌和规格后的剩余部分作为品名
            processed_name = orig_name
            
            # 去除品牌
            brand = result_df.at[idx, 'brand']
            if brand and brand in processed_name:
                processed_name = processed_name.replace(brand, '', 1)
                
            # 去除规格
            specification = result_df.at[idx, 'specification']
            if specification and specification in processed_name:
                processed_name = processed_name.replace(specification, '')
                
            # 清理和标准化
            processed_name = re.sub(r'[%@#&:;.,\*]+', ' ', processed_name).strip()
            processed_name = re.sub(r'\s+', ' ', processed_name).strip()
            
            if processed_name:
                name = processed_name
            else:
                # 如果仍然为空，使用原名称的一部分
                if len(parts) > 1:
                    # 尝试使用中间部分作为名称
                    middle_parts = parts[1:-1] if len(parts) > 2 else [parts[-1]]
                    name = ' '.join(middle_parts)
                else:
                    name = orig_name
            
            confidence -= 15
            applied_rules.append("used_fallback_name")
        
        # 8. 检测名称是否过短
        if len(name) < 2:
            confidence -= 20
            applied_rules.append("name_too_short")
        
        # 9. 评估是否符合标准格式：品牌+品名+规格
        if brand_found and spec_match and len(name) >= 2:
            confidence += 15
            applied_rules.append("standard_format")
        
        # 10. 设置清洗后的商品名称和置信度
        result_df.at[idx, 'cleaned_name'] = name
        # 限制置信度在0-1之间 (原来是0-100)
        result_df.at[idx, 'confidence_score'] = max(0, min(100, confidence)) / 100.0
        
        # 调试信息
        if idx < 5 or idx % 100 == 0:  # 记录前5行和每100行的一行
            logger.info(f"处理商品 {idx}: 原名称='{orig_name}', 清洗后='{name}', "
                        f"品牌='{result_df.at[idx, 'brand']}', 规格='{result_df.at[idx, 'specification']}', "
                        f"单位='{spec_unit}', 价格={price}, 标准单位='{standard_unit}', 标准价格={standard_price}, "
                        f"置信度={result_df.at[idx, 'confidence_score']}, 应用规则={applied_rules}")
    
    return result_df

def process_clean_task(task_id, db: Session = None):
    """
    处理数据清洗任务的后台任务
    
    Args:
        task_id: 数据清洗任务的ID
        db: 数据库会话
    """
    # 获取任务
    try:
        # 如果是通过数据库会话查询任务
        if db is not None:
            task = db.query(DataCleanTask).filter(DataCleanTask.id == task_id).first()
            if not task:
                logger.error(f"任务ID {task_id} 不存在")
                return "error"
            
            # 更新任务状态为处理中
            logger.info(f"开始处理任务 {task_id}: {task.name}")
            task.status = 'processing'
            db.commit()
            
            # 读取Excel文件
            logger.info(f"读取文件: {task.file_path}")
            df = pd.read_excel(task.file_path)
            
            # 检查必要的列是否存在
            if '商品名称' not in df.columns:
                error_msg = "数据文件必须包含'商品名称'列"
                task.status = 'failed'
                task.error_message = error_msg
                db.commit()
                logger.error(error_msg)
                return "error"
            
            # 清洗数据 - 使用混合策略，只对低置信度数据使用大模型
            use_llm = False
            llm_threshold = 0.9  # 置信度阈值设为0.9，表示置信度低于90%的数据会用大模型处理
            force_llm = False    # 不强制使用大模型处理所有数据
            
            # 如果任务有额外配置，可以覆盖上面的设置
            if hasattr(task, 'extra_config') and task.extra_config:
                try:
                    extra_config = json.loads(task.extra_config)
                    
                    # 读取配置并覆盖默认设置
                    if 'use_llm' in extra_config:
                        use_llm = extra_config.get('use_llm')
                    
                    if 'llm_threshold' in extra_config:
                        llm_threshold = float(extra_config.get('llm_threshold', 0.9))
                        # 确保阈值在有效范围内
                        llm_threshold = max(0.0, min(1.0, llm_threshold))
                    
                    if 'force_llm' in extra_config:
                        force_llm = extra_config.get('force_llm')
                    
                    logger.info(f"从任务配置中读取数据清洗设置: use_llm={use_llm}, threshold={llm_threshold}, force={force_llm}")
                except Exception as e:
                    logger.warning(f"解析任务额外配置失败: {str(e)}，将使用默认设置")
            
            # 清洗数据 - 使用混合策略
            logger.info(f"===== 开始数据清洗 =====")
            if use_llm:
                if force_llm:
                    logger.info(f"总共 {len(df)} 条记录，强制使用DeepSeek大模型处理所有数据")
                else:
                    logger.info(f"总共 {len(df)} 条记录，使用混合处理策略：先规则处理，对置信度<{llm_threshold:.2f}的数据使用大模型")
            else:
                logger.info(f"总共 {len(df)} 条记录，仅使用规则处理，不使用大模型")
            
            # 确保配置管理器已初始化并加载最新数据
            logger.info("确保配置管理器加载最新数据")
            config_manager.reload(db)
            logger.info(f"配置管理器已重新加载，当前品牌数量: {len(config_manager.brands)}")
            
            # 调用数据清洗函数，确保使用db参数
            result_df = clean_product_data(df, db, use_llm=use_llm, llm_threshold=llm_threshold, force_llm=force_llm)
            
            # 验证清洗后的数据
            logger.info(f"===== 清洗完成 =====")
            logger.info(f"结果数据有 {len(result_df)} 行，原始数据有 {len(df)} 行")
            if len(result_df) != len(df):
                logger.warning("清洗后的数据行数与原始数据不一致！")
                
            # 统计品牌和规格识别情况
            brands_identified = result_df['brand'].notna().sum()
            specs_identified = result_df['specification'].notna().sum()
            logger.info(f"品牌识别统计: 共识别出 {brands_identified}/{len(result_df)} 条商品的品牌")
            logger.info(f"规格识别统计: 共识别出 {specs_identified}/{len(result_df)} 条商品的规格")
            
            # 处理清洗过程中发现的新品牌和单位
            try:
                save_discovered_elements(db)
            except Exception as e:
                logger.warning(f"保存新发现的品牌和单位失败: {str(e)}")
                
            # 打印数据框信息
            logger.info(f"清洗结果数据框信息: \n列名: {result_df.columns.tolist()}\n"
                        f"前5行数据示例: \n{result_df.head(5)}")
            
            # 生成结果文件路径
            # 获取当前工作目录的绝对路径
            current_dir = os.path.abspath(os.getcwd())
            # 使用项目根目录的绝对路径
            project_root = os.path.abspath(os.path.join(current_dir, os.pardir)) if current_dir.endswith('backend') else current_dir
            
            # 在项目根目录下创建结果目录
            result_dir = os.path.join(project_root, 'results', 'data_clean')
            os.makedirs(result_dir, exist_ok=True)
            
            # 记录当前工作目录和项目根目录
            logger.info(f"当前工作目录: {current_dir}")
            logger.info(f"项目根目录: {project_root}")
            logger.info(f"结果目录绝对路径: {result_dir}")
            
            # 使用任务名称和ID生成唯一且清晰的文件名
            # 首先确定哪个名称字段有值
            task_display_name = task.task_name or task.name or f"task_{task_id}"
            safe_task_name = re.sub(r'[\\/*?:"<>|]', "_", task_display_name)  # 移除不安全的文件名字符
            timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
            result_filename = f"{safe_task_name}_id{task_id}_cleaned_{timestamp}.xlsx"
            
            # 使用绝对路径并统一使用正斜杠作为路径分隔符
            result_file = os.path.join(result_dir, result_filename).replace('\\', '/')
            
            # 记录绝对路径，方便调试
            logger.info(f"生成的结果文件绝对路径: {result_file}")
            
            # 保存结果 - 修复Excel保存方法
            try:
                logger.info(f"准备保存结果到Excel文件: {result_file}")
                # 检查文件目录是否存在
                result_dir = os.path.dirname(result_file)
                os.makedirs(result_dir, exist_ok=True)
                
                # 创建一个数据副本，并转换所有复杂对象为字符串
                export_df = result_df.copy()
                
                # 在每个单元格中查找复杂对象并转换它们
                for col in export_df.columns:
                    # 检查该列是否包含复杂对象
                    if export_df[col].apply(lambda x: isinstance(x, (dict, list, tuple, set))).any():
                        logger.info(f"列 '{col}' 包含复杂对象，将其转换为JSON字符串")
                        # 转换为JSON字符串
                        export_df[col] = export_df[col].apply(
                            lambda x: json.dumps(x, ensure_ascii=False) if isinstance(x, (dict, list, tuple, set)) else x
                        )
                
                # 完整日志记录DataFrame的信息
                logger.info(f"要导出的数据框信息:\n"
                            f"形状: {export_df.shape}\n"
                            f"列: {export_df.columns.tolist()}\n"
                            f"数据类型: {export_df.dtypes.to_dict()}")
                
                # 尝试两种不同的保存方法
                try:
                    # 方法1: 使用ExcelWriter和openpyxl引擎
                    with pd.ExcelWriter(result_file, engine='openpyxl') as writer:
                        export_df.to_excel(writer, sheet_name='清洗结果', index=False)
                    
                    # 验证文件是否成功创建
                    if not os.path.exists(result_file) or os.path.getsize(result_file) == 0:
                        raise Exception("使用openpyxl引擎保存失败，文件为空或不存在")
                except Exception as e1:
                    logger.warning(f"使用openpyxl引擎保存失败: {e1}，尝试使用xlsxwriter引擎")
                    
                    # 方法2: 使用xlsxwriter引擎
                    with pd.ExcelWriter(result_file, engine='xlsxwriter') as writer:
                        export_df.to_excel(writer, sheet_name='清洗结果', index=False)
                
                # 额外验证
                if not os.path.exists(result_file):
                    raise Exception("保存后文件不存在")
                
                file_size = os.path.getsize(result_file)
                if file_size == 0:
                    raise Exception("保存的文件大小为0")
                    
                logger.info(f"结果成功保存到: {result_file}, 文件大小: {file_size} 字节, 共 {len(result_df)} 行数据")
            except Exception as e:
                logger.error(f"保存Excel文件失败: {str(e)}")
                raise
            
            # 更新任务状态和结果文件路径（使用统一的正斜杠格式）
            task.status = 'completed'
            task.result_file_path = result_file  # 直接使用正斜杠格式的路径
            task.completed_at = datetime.now()
            db.commit()
            logger.info(f"任务 {task_id} 处理完成")
            return "completed"
            
        # 如果没有数据库会话，尝试使用备用API进行处理
        else:
            # 这里调用另一个函数实现，兼容无会话情况
            logger.info(f"无数据库会话，将使用备用API处理任务: {task_id}")
            return process_task_with_api(task_id)
            
    except Exception as e:
        error_message = f"处理任务 {task_id} 时出错: {str(e)}"
        logger.error(error_message)
        # 记录完整异常堆栈
        import traceback
        logger.error(f"异常堆栈: {traceback.format_exc()}")
        
        # 如果有数据库会话，更新任务状态
        if db is not None:
            try:
                task = db.query(DataCleanTask).filter(DataCleanTask.id == task_id).first()
                if task:
                    task.status = 'failed'
                    task.error_message = str(e)
                    db.commit()
            except Exception as db_error:
                logger.error(f"更新任务状态失败: {str(db_error)}")
        
        return "error"

def process_task_with_api(task_id: str) -> str:
    """
    通过API处理数据清洗任务（无数据库会话时使用）
    
    Args:
        task_id: 任务ID
        
    Returns:
        任务状态
    """
    logger.info(f"通过API处理任务: {task_id}")
    # 这里实现通过API获取任务信息并处理的逻辑
    # 具体实现需要根据系统API设计来完成
    
    # 示例实现
    try:
        # 这里应该有从API获取任务信息的逻辑
        task_info = get_task_info_from_api(task_id)
        
        # 读取文件
        file_path = task_info.get("file_path")
        df = pd.read_excel(file_path)
        
        # 验证文件格式
        if '商品名称' not in df.columns:
            logger.error(f"文件缺少必要的列: 商品名称")
            update_task_status_via_api(task_id, "error", "文件缺少必要的列: 商品名称")
            return "error"
        
        # 初始化配置
        config = get_config_from_api()
        
        # 处理数据
        # 这里需要一个不依赖db会话的数据处理函数
        result_df = process_data_without_db(df, config)
        
        # 保存结果
        result_path = f"./results/{task_id}_{datetime.now().strftime('%Y%m%d%H%M%S')}.xlsx"
        result_df.to_excel(result_path, index=False)
        
        # 更新任务状态
        update_task_status_via_api(task_id, "completed", result_file=result_path)
        
        return "completed"
        
    except Exception as e:
        logger.error(f"通过API处理任务失败: {str(e)}")
        update_task_status_via_api(task_id, "error", error_message=str(e))
        return "error"

# 辅助函数，实际使用时需要替换为真实的API调用
def get_task_info_from_api(task_id):
    # 这是一个示例，实际应该通过HTTP请求获取任务信息
    logger.warning("使用mock的任务信息API，实际应用中请替换为真实API")
    return {"task_id": task_id, "file_path": f"./data/{task_id}.xlsx"}

def get_config_from_api():
    # 这是一个示例，实际应该通过HTTP请求获取配置
    logger.warning("使用mock的配置API，实际应用中请替换为真实API")
    return {"brands": [], "units": []}

def update_task_status_via_api(task_id, status, error_message=None, result_file=None):
    # 这是一个示例，实际应该通过HTTP请求更新任务状态
    logger.warning("使用mock的任务状态更新API，实际应用中请替换为真实API")
    logger.info(f"更新任务 {task_id} 状态为 {status}")

def process_data_without_db(df, config):
    # 这是一个示例，实际应该实现一个不依赖db的数据处理函数
    logger.warning("使用mock的数据处理函数，实际应用中请替换为真实实现")
    return df

# 为了兼容性创建process_data_clean_task作为process_clean_task的别名
# process_data_clean_task = process_clean_task

def save_discovered_elements(db: Session):
    """
    保存清洗过程中发现的潜在新品牌和单位
    
    Args:
        db: 数据库会话
    """
    # 保存潜在的新品牌
    if hasattr(config_manager, 'potential_new_brands') and config_manager.potential_new_brands:
        logger.info(f"发现 {len(config_manager.potential_new_brands)} 个潜在新品牌")
        
        # 只保存出现频率较高的品牌（至少3次）
        frequent_brands = {brand: count for brand, count in config_manager.potential_new_brands.items() 
                           if count >= 3 and brand not in config_manager.brands}
        
        if frequent_brands:
            logger.info(f"尝试保存 {len(frequent_brands)} 个高频新品牌")
            for brand_name, count in frequent_brands.items():
                # 检查品牌是否已存在
                existing_brand = db.query(Brand).filter(Brand.name == brand_name).first()
                if not existing_brand:
                    # 创建新品牌记录
                    new_brand = Brand(
                        name=brand_name,
                        category='其他',  # 默认分类
                        is_active=False,  # 默认不激活，需管理员审核
                        confidence_score=min(count * 5, 80),  # 根据出现次数计算置信度
                        source='auto_discovery',
                        created_at=datetime.now()
                    )
                    db.add(new_brand)
                    logger.info(f"添加新品牌: {brand_name}，出现次数: {count}")
            
            try:
                db.commit()
                logger.info("新品牌保存成功")
            except Exception as e:
                db.rollback()
                logger.error(f"保存新品牌失败: {str(e)}")
    
    # 保存潜在的新单位
    if hasattr(config_manager, 'potential_new_units') and config_manager.potential_new_units:
        logger.info(f"发现 {len(config_manager.potential_new_units)} 个潜在新单位")
        
        # 只保存出现频率较高的单位（至少5次）
        frequent_units = {unit: count for unit, count in config_manager.potential_new_units.items() 
                          if count >= 5 and unit not in config_manager.units}
        
        if frequent_units:
            logger.info(f"尝试保存 {len(frequent_units)} 个高频新单位")
            for unit_name, count in frequent_units.items():
                # 检查单位是否已存在
                existing_unit = db.query(UnitConversion).filter(UnitConversion.source_unit == unit_name).first()
                if not existing_unit:
                    # 创建新单位记录（不设置转换公式，需管理员配置）
                    new_unit = UnitConversion(
                        source_unit=unit_name,
                        target_unit='piece',  # 默认目标单位
                        conversion_formula='x',  # 默认公式，需管理员修改
                        is_active=False,  # 默认不激活，需管理员审核
                        confidence_score=min(count * 5, 80),
                        source='auto_discovery',
                        created_at=datetime.now()
                    )
                    db.add(new_unit)
                    logger.info(f"添加新单位: {unit_name}，出现次数: {count}")
            
            try:
                db.commit()
                logger.info("新单位保存成功")
            except Exception as e:
                db.rollback()
                logger.error(f"保存新单位失败: {str(e)}")
    
    # 清除临时存储的潜在元素
    if hasattr(config_manager, 'potential_new_brands'):
        delattr(config_manager, 'potential_new_brands')
    if hasattr(config_manager, 'potential_new_units'):
        delattr(config_manager, 'potential_new_units')

def get_product_stats(df: pd.DataFrame, db: Session = None) -> dict:
    """
    获取商品数据的统计信息
    
    Args:
        df: 处理后的DataFrame
        db: 可选数据库会话，用于加载配置
        
    Returns:
        包含统计信息的字典
    """
    try:
        # 确保配置已加载
        if db and (not hasattr(config_manager, '_is_initialized') or not config_manager._is_initialized):
            config_manager.initialize(db)
        
        # 确保DataFrame包含所需列
        required_columns = ['brand', 'unit', 'brand_in_database', 'extract_method']
        for col in required_columns:
            if col not in df.columns:
                if col == 'brand_in_database':
                    df[col] = False
                elif col == 'extract_method':
                    df[col] = 'rule'  # 默认为规则提取
                else:
                    df[col] = ''
                
        # 处理空值
        df['brand'] = df['brand'].fillna('')
        df['unit'] = df['unit'].fillna('')
        df['brand_in_database'] = df['brand_in_database'].fillna(False)
        df['extract_method'] = df['extract_method'].fillna('rule')
        
        # 分析品牌数据
        brands_in_db = df[df['brand_in_database'] == True]['brand'].value_counts().to_dict()
        brands_not_in_db = df[(df['brand'] != '') & (df['brand_in_database'] == False)]['brand'].value_counts().to_dict()
        
        # 单位统计
        unit_counts = df[df['unit'] != '']['unit'].value_counts().to_dict()
        
 #       提取方式统计
        extract_method_counts = df['extract_method'].value_counts().to_dict()
        
        # 如果没有识别到任何品牌，添加一个默认值
        if not brands_in_db and not brands_not_in_db:
            brands_not_in_db = {'未识别': 0}
            
        # 如果没有识别到任何单位，添加一个默认值
        if not unit_counts:
            unit_counts = {'未识别': 0}
            
        # 统计数据
        stats = {
            "total_products": len(df),
            "identified_brands": sum(1 for b in df['brand'] if b),
            "brands_in_database": brands_in_db,
            "brands_not_in_database": brands_not_in_db,
            "units_distribution": unit_counts,
            "extract_method_distribution": extract_method_counts,
        }
        return stats
    except Exception as e:
        logger.error(f"生成统计信息失败: {str(e)}")
        # 返回空统计信息而不是抛出异常
        return {
            "total_products": len(df) if isinstance(df, pd.DataFrame) else 0,
            "identified_brands": 0,
            "brands_in_database": {},
            "brands_not_in_database": {},
            "units_distribution": {},
            "extract_method_distribution": {"error": 0},
            "error": str(e)
        }

def get_excel_preview(file_path: str, sheet_name: str = None) -> dict:
    """
    获取Excel文件的预览数据，返回表头和所有行数据
    
    Args:
        file_path: Excel文件路径
        sheet_name: 工作表名称，可选
        
    Returns:
        包含预览数据的字典
    """
    try:
        # 读取Excel文件
        if sheet_name:
            df = pd.read_excel(file_path, sheet_name=sheet_name)
        else:
            df = pd.read_excel(file_path)
        
        # 获取所有工作表名称
        excel_file = pd.ExcelFile(file_path)
        sheet_names = excel_file.sheet_names
        
        # 获取表头（列名）
        headers = df.columns.tolist()
        
        # 获取所有行数据，而不仅仅是前10行
        preview_data = df.to_dict(orient='records')
        
        # 数据类型转换为可序列化的格式
        for row in preview_data:
            for key, value in row.items():
                if pd.isna(value):
                    row[key] = None
                elif isinstance(value, pd.Timestamp):
                    # 确保时间戳格式正确且不包含换行符
                    try:
                        row[key] = value.strftime('%Y-%m-%dT%H:%M:%S')
                    except Exception as e:
                        logger.warning(f"时间戳格式化失败: {str(e)}")
                        row[key] = str(value).replace('\n', ' ')
        
        return {
            "sheet_names": sheet_names,
            "current_sheet": sheet_name or sheet_names[0],
            "headers": headers,
            "rows": preview_data,
            "total_rows": len(df)
        }
    except Exception as e:
        logger.error(f"获取Excel预览失败: {str(e)}")
        return {
            "sheet_names": [],
            "current_sheet": sheet_name,
            "headers": [],
            "rows": [],
            "total_rows": 0,
            "error": str(e)
        }

# 新增大模型处理相关函数
# def call_deepseek_api(prompt):
#     """调用DeepSeek API
    
#     Args:
#         prompt: 提示词
        
#     Returns:
#         API响应
#     """
#     import os
#     import time
    
#     try:
#         # 动态导入OpenAI SDK，避免强依赖
#         from openai import OpenAI
#     except ImportError:
#         logger.error("OpenAI SDK未安装，请运行: pip install openai")
#         try:
#             # 尝试自动安装
#             import subprocess
#             logger.info("尝试自动安装OpenAI SDK...")
#             subprocess.check_call(["pip", "install", "openai"])
#             from openai import OpenAI
#             logger.info("OpenAI SDK安装成功")
#         except Exception as e:
#             logger.error(f"自动安装OpenAI SDK失败: {str(e)}")
#             raise ImportError("请手动安装OpenAI SDK: pip install openai")
    
#     # 获取API密钥
#     api_key = None
    
#     # 优先级1: 从配置管理器获取
#     try:
#         if hasattr(config_manager, '_is_initialized') and config_manager._is_initialized:
#             if hasattr(config_manager, '_system_config'):
#                 api_key = config_manager._system_config.get("DEEPSEEK_API_KEY")
#                 if api_key:
#                     logger.info("从配置管理器获取DeepSeek API密钥成功")
#     except Exception as e:
#         logger.warning(f"从配置管理器获取密钥失败: {str(e)}")
    
#     # 优先级2: 从环境变量获取
#     if not api_key:
#         try:
#             api_key = os.environ.get("DEEPSEEK_API_KEY")
#             if api_key:
#                 logger.info("使用环境变量中的DeepSeek密钥")
#         except Exception as e:
#             logger.warning(f"从环境变量获取密钥失败: {str(e)}")
    
#     # 优先级3: 从.env文件获取（仅用于开发测试）
#     if not api_key:
#         try:
#             env_path = os.path.join(os.getcwd(), '.env')
#             if os.path.exists(env_path):
#                 with open(env_path, 'r') as f:
#                     for line in f:
#                         if line.startswith('DEEPSEEK_API_KEY='):
#                             api_key = line.strip().split('=', 1)[1].strip('"\'')
#                             if api_key:
#                                 logger.info("使用.env文件中的DeepSeek密钥")
#                                 break
#         except Exception as e:
#             logger.warning(f"从.env文件获取密钥失败: {str(e)}")
    
#     # 最终检查
#     if not api_key:
#         error_msg = "未能获取DEEPSEEK_API_KEY，请在配置管理器、环境变量或.env文件中设置"
#         logger.error(error_msg)
#         raise ValueError(error_msg)
    
#     # 获取其他API设置
#     base_url = "https://api.deepseek.com"  # 默认URL
#     model = "deepseek-chat"  # 默认模型
#     temperature = 0.1  # 默认温度
#     max_tokens = 1024  # 默认最大token数 - 使用适当的默认值
    
#     # 尝试从配置管理器获取更多设置
#     try:
#         if hasattr(config_manager, '_is_initialized') and config_manager._is_initialized:
#             system_config = config_manager._system_config
            
#             # API URL
#             if "DEEPSEEK_API_URL" in system_config:
#                 base_url = system_config.get("DEEPSEEK_API_URL", base_url)
            
#             # 模型名称
#             if "DEEPSEEK_MODEL" in system_config:
#                 model = system_config.get("DEEPSEEK_MODEL", model)
            
#             # 温度处理
#             try:
#                 temp_value = system_config.get("DEEPSEEK_TEMPERATURE")
#                 if temp_value is not None:
#                     temperature = float(temp_value)
#                     # 确保温度在合理范围内
#                     temperature = max(0.0, min(1.0, temperature))
#             except Exception as e:
#                 logger.warning(f"温度值解析错误，使用默认值0.1: {str(e)}")
            
#             # max_tokens处理
#             try:
#                 tokens_value = system_config.get("DEEPSEEK_MAX_TOKENS")
#                 if tokens_value is not None:
#                     max_tokens_parsed = int(tokens_value)
#                     # 确保max_tokens在DeepSeek API允许范围内 (1-8192)
#                     max_tokens = max(1, min(8192, max_tokens_parsed))
#                     logger.info(f"从配置获取max_tokens值: {max_tokens}")
#             except Exception as e:
#                 logger.warning(f"max_tokens值解析错误，使用默认值1024: {str(e)}")
#     except Exception as e:
#         logger.warning(f"从配置管理器获取API设置失败: {str(e)}")
    
#     # 尝试从环境变量获取更多设置
#     env_api_url = os.environ.get("DEEPSEEK_API_URL")
#     if env_api_url:
#         base_url = env_api_url
#         logger.info(f"从环境变量获取API URL: {base_url}")
    
#     env_model = os.environ.get("DEEPSEEK_MODEL")
#     if env_model:
#         model = env_model
#         logger.info(f"从环境变量获取模型名称: {model}")
    
#     try:
#         temp_env = os.environ.get("DEEPSEEK_TEMPERATURE")
#         if temp_env:
#             temperature = float(temp_env)
#             temperature = max(0.0, min(1.0, temperature))
#             logger.info(f"从环境变量获取温度值: {temperature}")
#     except Exception as e:
#         logger.warning(f"解析环境变量温度值失败: {str(e)}")
        
#     try:
#         tokens_env = os.environ.get("DEEPSEEK_MAX_TOKENS")
#         if tokens_env:
#             try:
#                 max_tokens_parsed = int(tokens_env)
#                 # 确保max_tokens在DeepSeek API允许范围内 (1-8192)
#                 max_tokens = max(1, min(8192, max_tokens_parsed))
#                 logger.info(f"从环境变量获取max_tokens值: {max_tokens}")
#             except Exception:
#                 pass
#     except Exception as e:
#         logger.warning(f"解析环境变量max_tokens值失败: {str(e)}")
    
#     # 最终配置汇总
#     logger.info(f"最终DeepSeek API配置 - 基础URL: {base_url}, 模型: {model}, 温度: {temperature}, 最大token数: {max_tokens}")
#     logger.info(f"请求提示词长度: {len(prompt)} 字符")
#     logger.info(f"请求提示词前100字符: {prompt[:100]}...")
    
#     # 使用OpenAI SDK调用API
#     max_retries = 3
#     current_retry = 0
#     retry_delay = 5
    
#     while current_retry < max_retries:
#         try:
#             logger.info(f"==== 调用DeepSeek API开始 (尝试 {current_retry+1}/{max_retries}) ====")
            
#             # 检查环境变量中是否存在代理配置
#             http_proxy = os.environ.get("HTTP_PROXY")
#             https_proxy = os.environ.get("HTTPS_PROXY")
#             if http_proxy or https_proxy:
#                 logger.info(f"检测到代理设置: HTTP_PROXY={http_proxy}, HTTPS_PROXY={https_proxy}")
#                 logger.warning("OpenAI SDK不支持直接传入proxies参数，将通过环境变量设置")
            
#             # 初始化OpenAI客户端
#             client = OpenAI(
#                 api_key=api_key,
#                 base_url=base_url
#             )
            
#             # 调用API
#             response = client.chat.completions.create(
#                 model=model,
#                 messages=[
#                     {"role": "system", "content": "你是一位专业的商品数据清洗和标准化专家，擅长从商品描述中提取结构化信息。"},
#                     {"role": "user", "content": prompt}
#                 ],
#                 temperature=temperature,
#                 max_tokens=max_tokens,
#                 stream=False
#             )
            
#             # 转换为字典格式，与旧API保持一致
#             response_dict = {
#                 "id": response.id,
#                 "object": response.object,
#                 "created": response.created,
#                 "model": response.model,
#                 "choices": [
#                     {
#                         "index": choice.index,
#                         "message": {
#                             "role": choice.message.role,
#                             "content": choice.message.content
#                         },
#                         "finish_reason": choice.finish_reason
#                     } for choice in response.choices
#                 ]
#             }
            
#             # 记录令牌使用情况
#             if hasattr(response, 'usage'):
#                 response_dict["usage"] = {
#                     "prompt_tokens": response.usage.prompt_tokens,
#                     "completion_tokens": response.usage.completion_tokens,
#                     "total_tokens": response.usage.total_tokens
#                 }
#                 logger.info(f"DeepSeek API令牌使用情况: 提示词={response.usage.prompt_tokens}, "
#                            f"完成={response.usage.completion_tokens}, "
#                            f"总计={response.usage.total_tokens}")
            
#             # 记录响应内容
#             if response.choices and len(response.choices) > 0:
#                 content = response.choices[0].message.content
#                 logger.info(f"DeepSeek API响应内容长度: {len(content)} 字符")
#                 logger.info(f"DeepSeek API响应内容前100字符: {content[:100]}...")
#             else:
#                 logger.warning("DeepSeek API响应中没有找到有效内容")
            
#             logger.info(f"==== 调用DeepSeek API结束 ====")
#             return response_dict
            
#         except Exception as e:
#             logger.error(f"调用DeepSeek API失败: {str(e)}")
#             current_retry += 1
            
#             if current_retry < max_retries:
#                 retry_wait = retry_delay * (2 ** (current_retry - 1))  # 指数退避策略
#                 logger.warning(f"等待 {retry_wait} 秒后重试...")
#                 time.sleep(retry_wait)
#             else:
#                 logger.error(f"DeepSeek API在 {max_retries} 次尝试后仍然失败")
#                 raise Exception(f"DeepSeek API调用失败: {str(e)}")

def parse_deepseek_response(response):
    """解析DeepSeek API响应
    
    Args:
        response: API响应
        
    Returns:
        解析后的结果列表
    """
    try:
        logger.info("开始解析DeepSeek API响应...")
        
        # 检查响应格式
        if not isinstance(response, dict):
            logger.error(f"响应格式不正确，预期为字典，实际为: {type(response)}")
            return []
        
        # 检查响应中是否包含choices
        if 'choices' not in response or not response['choices']:
            logger.error(f"响应中未找到choices字段: {json.dumps(response, ensure_ascii=False)[:500]}...")
            return []
        
        # 提取响应中的内容
        content = response.get('choices', [{}])[0].get('message', {}).get('content', '')
        
        # 检查内容是否为空
        if not content.strip():
            logger.error("响应中content字段为空")
            return []
        
        # 记录完整响应内容用于调试
        logger.info(f"响应内容长度: {len(content)} 字符")
        logger.info(f"响应内容前500字符: {content[:500]}...")
        
        # 提取JSON部分
        json_start = content.find('[')
        json_end = content.rfind(']') + 1
        
        if json_start == -1 or json_end == 0:
            logger.error(f"无法从响应中提取JSON: 未找到JSON数组标记")
            # 尝试查找可能的JSON对象
            json_start = content.find('{')
            json_end = content.rfind('}') + 1
            if json_start == -1 or json_end == 0:
                logger.error("也未找到JSON对象标记，无法解析")
                return []
            logger.info(f"找到可能的JSON对象，尝试解析...")
        
        # 提取JSON字符串
        json_str = content[json_start:json_end]
        logger.info(f"提取的JSON长度: {len(json_str)} 字符")
        logger.info(f"JSON前5000字符: {json_str[:5000]}...")
        
        # 解析JSON
        try:
            result = json.loads(json_str)
            logger.info(f"JSON解析成功")
            
            # 将单个对象转换为列表
            if isinstance(result, dict):
                logger.info("响应为单个对象，转换为列表")
                result = [result]
            
            logger.info(f"成功解析JSON，包含 {len(result)} 条商品数据")
            
            # 记录数据样例
            if result and len(result) > 0:
                logger.info(f"第一条数据示例: {json.dumps(result[0], ensure_ascii=False)}")
                # 检查关键字段
                required_fields = ['original_name', 'brand', 'spec_value', 'spec_unit']
                missing_fields = [field for field in required_fields if field not in result[0]]
                if missing_fields:
                    logger.warning(f"数据缺少关键字段: {', '.join(missing_fields)}")
            
            return result
        except json.JSONDecodeError as e:
            logger.error(f"JSON解析错误: {str(e)}")
            logger.error(f"问题的JSON字符串: {json_str[:200]}...")
            return []
    except Exception as e:
        logger.error(f"解析DeepSeek响应失败: {str(e)}")
        import traceback
        logger.error(f"异常堆栈: {traceback.format_exc()}")
        logger.error(f"原始响应内容: {str(response)[:1000]}...")
        return []

def deepseek_data_clean(product_data, db=None, batch_size=50):
    """调用DeepSeek API进行数据清洗
    
    Args:
        product_data: 商品数据列表
        db: 数据库会话，用于初始化配置
        batch_size: 批处理大小
        
    Returns:
        处理后的结果
    """
    results = []
    
    # 确保配置管理器已初始化，并且正确加载品牌信息
    if db:
        try:
            logger.info("重新初始化配置管理器以确保数据最新")
            config_manager.initialize(db)
            
            # 验证品牌数据是否加载成功
            brands_count = len(config_manager.brands)
            units_count = len(config_manager.units)
            logger.info(f"配置管理器初始化完成，加载了 {brands_count} 个品牌和 {units_count} 个单位")
            
            # 记录前10个品牌用于调试
            if brands_count > 0:
                sample_brands = config_manager.brands[:100]
                logger.info(f"品牌样例: {', '.join(sample_brands)}")
            else:
                logger.warning("没有加载到任何品牌信息，请检查数据库")
        except Exception as e:
            logger.error(f"配置管理器初始化失败: {str(e)}")
            import traceback
            logger.error(f"异常详情: {traceback.format_exc()}")
    else:
        logger.warning("未提供数据库会话，配置管理器可能无法获取最新数据")
    
    # 批量处理数据
    batches = [product_data[i:i+batch_size] for i in range(0, len(product_data), batch_size)]
    logger.info(f"将 {len(product_data)} 条数据分成 {len(batches)} 个批次进行处理")
    
    for i, batch in enumerate(batches):
        logger.info(f"===== 开始处理批次 {i+1}/{len(batches)}, 包含 {len(batch)} 条数据 =====")
        
        try:
            # 使用新的提示词生成函数生成提示词
            logger.info(f"为批次 {i+1} 生成提示词...")
            prompt = generate_deepseek_prompt(batch, config_manager)
            logger.info(f"提示词生成完成，长度: {len(prompt)} 字符")
            
            # 记录品牌和单位统计
            brands_count = len(config_manager.brands)
            units_count = len(config_manager.units)
            logger.info(f"当前配置包含 {brands_count} 个品牌和 {units_count} 个单位")
            
            # 调用API
            logger.info(f"调用DeepSeek API处理批次 {i+1}...")
            # 使用导入的模块函数，传入config_manager参数
            from backend.data_clean.api_service import call_deepseek_api as api_call_deepseek
            response = api_call_deepseek(prompt, config_manager=config_manager)
            
            # 记录API响应状态
            logger.info(f"DeepSeek API响应成功，开始解析结果...")
            
            # 解析结果 - 只解析一次响应
            logger.info(f"解析DeepSeek API响应...")
            parsed_results = parse_deepseek_response(response)
            
            if parsed_results:
                logger.info(f"批次 {i+1} 成功解析 {len(parsed_results)} 条结果")
                
                # 检查品牌识别情况
                brands_identified = sum(1 for item in parsed_results if item.get('brand'))
                logger.info(f"批次 {i+1} 识别到品牌的数据: {brands_identified}/{len(parsed_results)}")
                
                # 检查规格提取情况
                specs_identified = sum(1 for item in parsed_results if item.get('spec_value') and item.get('spec_unit'))
                logger.info(f"批次 {i+1} 识别到规格的数据: {specs_identified}/{len(parsed_results)}")
                
                # 添加详细日志记录第一个结果的品牌字段
                if parsed_results and len(parsed_results) > 0:
                    first_result = parsed_results[0]
                    brand = first_result.get('brand', '')
                    logger.info(f"第一个结果品牌字段: '{brand}'")
                    
                    # 检查brands_list是否有效
                    if brand and config_manager and hasattr(config_manager, 'brands'):
                        brand_in_list = brand in config_manager.brands
                        logger.info(f"品牌'{brand}'在配置管理器的品牌列表中: {brand_in_list}")
                
                results.extend(parsed_results)
            else:
                logger.warning(f"批次 {i+1} 未返回任何有效结果")
                # 记录更详细的错误信息
                logger.error(f"分析响应内容: {str(response)[:500]}...")
            
            # 在批次之间添加延迟，避免API限流
            if i < len(batches) - 1:  # 如果不是最后一个批次
                delay_seconds = 5  # 设置延迟时间为5秒
                logger.info(f"批次处理完成，等待 {delay_seconds} 秒后处理下一批次...")
                import time
                time.sleep(delay_seconds)
        except Exception as e:
            logger.error(f"处理批次 {i+1} 时出错: {str(e)}")
            # 打印完整的异常堆栈跟踪
            import traceback
            logger.error(f"异常堆栈跟踪: {traceback.format_exc()}")
    
    logger.info(f"所有批次处理完成，总共获取 {len(results)} 条结果")
    return results

def process_deepseek_results(raw_results, db):
    """处理大模型返回结果并验证标准化
    
    Args:
        raw_results: 大模型API返回的结果
        db: 数据库会话
        
    Returns:
        验证后的DataFrame
    """
    # 确保配置管理器已初始化
    if not hasattr(config_manager, '_is_initialized') or not config_manager._is_initialized:
        config_manager.initialize(db)
    
    logger.info(f"开始处理DeepSeek返回的 {len(raw_results)} 条结果")
    
    # 转换为DataFrame
    result_df = pd.DataFrame(raw_results)
    
    # 记录列名，用于调试
    logger.info(f"DeepSeek返回的列名: {result_df.columns.tolist()}")
    
    # 添加缺失的列
    required_columns = [
        'original_name', 'original_price', 'brand', 'brand_in_database', 
        'cleaned_name', 'spec_value', 'spec_unit', 'standard_unit', 
        'standard_value', 'standard_price'
    ]
    
    for col in required_columns:
        if col not in result_df.columns:
            if col in ['original_price', 'spec_value', 'standard_value', 'standard_price']:
                result_df[col] = 0.0
                logger.info(f"添加缺失的数值列: {col}")
            else:
                result_df[col] = ''
                logger.info(f"添加缺失的文本列: {col}")
    
    # 验证并修正大模型输出的标准单位和价格计算
    brands_found = 0
    for idx, row in result_df.iterrows():
        try:
            # 验证品牌是否在数据库中
            if row['brand'] and isinstance(row['brand'], str):
                # 规范化品牌名称 - 去除多余空格
                brand_name = row['brand'].strip()
                result_df.at[idx, 'brand'] = brand_name
                
                # 检查品牌是否在数据库中
                is_in_db = brand_name in config_manager.brands
                result_df.at[idx, 'brand_in_database'] = is_in_db
                if is_in_db:
                    brands_found += 1
                    
                # 记录品牌识别结果
                if idx < 5 or idx % 100 == 0:
                    logger.info(f"商品 {idx}: 品牌='{brand_name}', 在数据库中={is_in_db}")
            
            # 处理规格值和单位转换
            if pd.notna(row['spec_value']) and pd.notna(row['spec_unit']):
                spec_value = row['spec_value']
                spec_unit = row['spec_unit'] if isinstance(row['spec_unit'], str) else str(row['spec_unit'])
                
                # 处理范围值，如"4.5-5斤"
                spec_text = str(spec_value) + spec_unit
                if isinstance(spec_value, str) and '-' in spec_value:
                    # 提取范围的最小值作为计算标准
                    range_match = re.match(r'(\d+\.?\d*)[-~](\d+\.?\d*)', spec_value)
                    if range_match:
                        min_value = float(range_match.group(1))
                        # 更新规格值为最小值
                        spec_value = min_value
                        result_df.at[idx, 'spec_value'] = min_value
                        
                        # 记录范围规格处理
                        logger.info(f"处理范围规格 {idx}: 原始值='{row['spec_value']}', 使用最小值={min_value}")
                else:
                    # 确保规格值是浮点数
                    try:
                        spec_value = float(spec_value)
                        result_df.at[idx, 'spec_value'] = spec_value
                    except (ValueError, TypeError):
                        logger.warning(f"无法转换规格值为浮点数: {spec_value}")
                        continue
                
                # 保存完整规格文本
                result_df.at[idx, 'specification'] = spec_text
                result_df.at[idx, 'unit'] = spec_unit
                
                # 单位转换和标准价格计算
                if spec_unit in config_manager.unit_conversion:
                    try:
                        # 使用配置中的正确转换规则重新计算
                        standard_value = config_manager.unit_conversion[spec_unit](spec_value)
                        standard_unit = config_manager.standard_unit_map.get(spec_unit, spec_unit)
                        
                        # 更新标准值和单位
                        result_df.at[idx, 'standard_value'] = standard_value
                        result_df.at[idx, 'standard_unit'] = standard_unit
                        
                        # 重新计算标准价格
                        if pd.notna(row['original_price']) and float(row['original_price']) > 0 and standard_value > 0:
                            standard_price = float(row['original_price']) / standard_value
                            result_df.at[idx, 'standard_price'] = round(standard_price, 2)
                            
                            # 记录单位转换和价格计算
                            if idx < 5 or idx % 100 == 0:
                                logger.info(f"单位转换 {idx}: {spec_value}{spec_unit} -> {standard_value}{standard_unit}, "
                                           f"价格: {row['original_price']} -> {round(standard_price, 2)}/每{standard_unit}")
                    except Exception as e:
                        logger.warning(f"单位转换失败 {idx}: {str(e)}")
        except Exception as e:
            logger.warning(f"处理行 {idx} 失败: {str(e)}")
    
    logger.info(f"品牌识别统计: 总计 {len(result_df)} 条数据，成功识别品牌 {brands_found} 条")
    
    # 重命名和调整列以匹配原始结果格式
    column_mapping = {
        'original_name': '商品名称',
        'original_price': '商品价格',
        'brand': 'brand',
        'brand_in_database': 'brand_in_database',
        'cleaned_name': 'cleaned_name',
        'spec_value': 'spec_value',
        'spec_unit': 'unit',
        'standard_unit': 'standard_unit',
        'standard_value': 'standard_value',
        'standard_price': 'standard_price'
    }
    
    # 重命名列
    result_df = result_df.rename(columns=column_mapping)
    logger.info(f"重命名列名完成: {result_df.columns.tolist()}")
    
    # 添加置信度分数列
    result_df['confidence_score'] = 0.85  # 默认给大模型处理的结果一个较高的置信度(0.85)
    
    # 添加提取方式列
    result_df['extract_method'] = 'ai'
    
    # 添加规格列（如果没有被前面的代码添加）
    if 'specification' not in result_df.columns:
        result_df['specification'] = result_df.apply(
            lambda row: f"{row['spec_value']}{row['unit']}" if pd.notna(row['spec_value']) and pd.notna(row['unit']) else "",
            axis=1
        )
    
    logger.info(f"DeepSeek结果处理完成，最终数据框包含 {len(result_df)} 行，{len(result_df.columns)} 列")
    return result_df

def hybrid_data_cleaning(df: pd.DataFrame, db: Session = None, use_model_threshold: float = 0.9, force_model: bool = False, batch_size: int = 100) -> pd.DataFrame:
    """
    混合数据清洗策略：先用规则方法处理所有数据，对于置信度低的数据使用大模型批量处理
    
    Args:
        df: 原始DataFrame
        db: 数据库会话
        use_model_threshold: 使用模型处理的置信度阈值，默认0.9表示低于90%置信度的数据用大模型处理
        force_model: 强制使用模型处理所有数据，默认False
        batch_size: 批处理大小，默认100条
        
    Returns:
        处理后的DataFrame
    """
    start_time = time.time()
    total_records = len(df)
    logger.info(f"开始混合数据清洗，数据量: {total_records}，置信度阈值: {use_model_threshold}，强制使用大模型: {force_model}")
    
    # 确保配置管理器已初始化
    if not hasattr(config_manager, '_is_initialized') or not config_manager._is_initialized:
        if db is not None:
            logger.info("使用数据库会话初始化配置管理器")
            config_manager.initialize(db)
        else:
            logger.warning("没有提供数据库会话，尝试直接初始化配置管理器")
            config_manager.initialize()
    
    # 步骤1：先用规则方法处理所有数据
    logger.info("第一步：使用规则方法处理所有数据")
    rule_results_df = ai_driven_data_cleaning(df, db)
    
    # 如果不使用大模型，直接返回规则处理结果
    if (not force_model and use_model_threshold <= 0):
        logger.info("不使用大模型，直接返回规则处理结果")
        rule_results_df['process_method'] = 'rule'  # 标记处理方法
        return rule_results_df
    
    # 步骤2：根据置信度阈值筛选需要大模型处理的数据
    if force_model:
        # 强制所有数据都用大模型处理
        logger.info("强制使用大模型处理所有数据")
        low_confidence_df = df.copy()
        low_confidence_indices = df.index.tolist()
    else:
        # 只对置信度低于阈值的数据使用大模型处理
        low_confidence_mask = rule_results_df['confidence_score'] < use_model_threshold
        low_confidence_indices = rule_results_df.index[low_confidence_mask].tolist()
        low_confidence_df = df.loc[low_confidence_indices].copy()
        logger.info(f"根据置信度阈值 {use_model_threshold} 筛选出 {len(low_confidence_df)} 条数据需要使用大模型处理 (总数据: {total_records})")
    
    # 如果没有需要处理的低置信度数据，直接返回规则处理结果
    if len(low_confidence_df) == 0:
        logger.info("没有需要大模型处理的数据，直接返回规则处理结果")
        rule_results_df['process_method'] = 'rule'  # 标记处理方法
        return rule_results_df
    
    # 步骤3：使用大模型处理低置信度数据
    logger.info(f"第三步：使用大模型批量处理 {len(low_confidence_df)} 条低置信度数据，批量大小: {batch_size}")
    try:
        # 准备待处理数据
        product_data = prepare_product_data(low_confidence_df)
        
        # 使用大模型处理
        model_results = deepseek_data_clean(product_data, db=db, batch_size=batch_size)
        
        # 处理大模型返回结果
        if model_results:
            model_df = process_deepseek_results(model_results, db)
            logger.info(f"大模型处理完成，获得 {len(model_df)} 条结果")
        else:
            logger.warning("大模型处理未返回有效结果，将保留规则处理结果")
            rule_results_df['process_method'] = 'rule'  # 标记处理方法
            return rule_results_df
        
        # 步骤4：合并结果 - 用大模型处理结果替换对应的规则处理结果
        final_df = rule_results_df.copy()
        
        # 添加处理方法标记
        if 'process_method' not in final_df.columns:
            final_df['process_method'] = 'rule'
            
        # 检查模型处理结果与低置信度数据索引的匹配情况
        if len(model_df) == len(low_confidence_indices):
            # 索引数量匹配，安全更新
            logger.info("模型处理结果数量与待处理数据匹配，进行替换")
            
            # 设置模型处理结果的索引与原始数据相同
            model_df.index = low_confidence_indices
            
            # 更新最终结果中的相应列
            for col in model_df.columns:
                if col in final_df.columns:
                    final_df.loc[low_confidence_indices, col] = model_df[col].values
        else:
            # 索引数量不匹配，使用安全更新方式逐行处理
            logger.warning(f"模型处理结果数量与待处理数据不匹配，模型结果: {len(model_df)}，原始数据: {len(low_confidence_indices)}")
            available_results = min(len(model_df), len(low_confidence_indices))
            
            for i in range(available_results):
                idx = low_confidence_indices[i]
                for col in model_df.columns:
                    if col in final_df.columns:
                        final_df.at[idx, col] = model_df.iloc[i][col]
        
        # 标记处理方法
        final_df.loc[low_confidence_indices[:len(model_df)], 'process_method'] = 'model'
        
        # 步骤5：记录处理结果统计
        rule_count = sum(final_df['process_method'] == 'rule')
        model_count = sum(final_df['process_method'] == 'model')
        
        logger.info(f"混合数据清洗完成，耗时: {time.time() - start_time:.2f}秒")
        logger.info(f"处理结果统计: 总数据={len(final_df)}，规则处理={rule_count}条 ({rule_count/total_records:.1%})，"
                    f"模型处理={model_count}条 ({model_count/total_records:.1%})")
        
        return final_df
        
    except Exception as e:
        # 出错时返回规则处理结果
        logger.error(f"大模型处理过程中出错: {str(e)}")
        import traceback
        logger.error(f"异常堆栈: {traceback.format_exc()}")
        
        logger.warning("由于大模型处理失败，返回规则处理结果")
        rule_results_df['process_method'] = 'rule'  # 标记处理方法
        return rule_results_df

# 处理任务
# logger.info(f"开始处理清洗任务 {task_id}")

# 先加载配置 - 确保单位转换管理的配置已加载
# config_manager.reload(db)
# logger.info(f"配置信息单位转换: {config_manager.units}") 

# 基于我的深入分析和开发的解决方案，我已经创建了一套增强的数据清洗系统，重点解决了品牌识别和商品信息提取的问题。以下是解决方案概要：
# 主要增强模块
# 品牌识别增强器 (enhanced_brand_recognition.py)
# 使用模糊匹配技术处理拼写错误的品牌名
# 应用jieba分词提取候选品牌片段
# 支持spaCy命名实体识别识别未知品牌
# 基于匹配置信度计算品牌识别的可靠性
# 规格提取器 (enhanced_specification_extractor.py)
# 处理多种规格格式：标准规格、约数规格、范围规格、复合规格等
# 优先级处理机制，如"750ml每瓶"时优先使用"瓶"作为单位
# 复合规格智能处理，如"500ml6瓶整箱"
# 自动转换为标准单位和计算标准价格
# 文本清洗器 (text_cleaner.py)
# 高效过滤"新老包装随机发"、"顺丰"等无关信息
# 识别并移除括号内的无关内容
# 使用jieba分词和TF-IDF提取商品的核心信息
# 智能保留对商品描述重要的关键词
# 增强处理器 (enhanced_processor.py)
# 整合上述所有增强模块
# 优化处理顺序：预处理→规格提取→品牌识别→去除无关信息→品名提取
# 多维度置信度评分系统
# 混合处理策略：规则处理+大模型处理
# 处理流程优化
# 优化的处理顺序
# 先进行规格提取，因为规格通常有明确的数字+单位模式，最容易识别
# 然后识别品牌，基于已知品牌列表和模糊匹配
# 接着移除无关信息，包括营销词、配送信息等
# 最后提取核心品名，即剩余的重要信息
# 增强的混合处理策略
# 首先使用增强规则处理所有数据，这是高效且可控的
# 对于置信度低的项目，使用大模型进行深度处理
# 智能合并规则处理和模型处理的结果
# 系统集成
# 与现有系统集成
# 修改了原有processor.py，使其自动检测并使用增强功能
# 保持向后兼容，若增强功能不可用则回退到原有处理方法
# 提供了安装脚本和批处理文件，方便部署
# 依赖管理
# 创建install_nlp_dependencies.py安装所需NLP库
# 优雅降级机制：若某些NLP库不可用，相应功能会被禁用但不影响整体功能
# 主要优势
# 更强大的品牌识别能力
# 能处理品牌名的变体和拼写错误
# 可以从复合品牌中（如"多点工坊×马大姐"）识别已知品牌
# 全面的规格处理
# 支持多种规格格式和单位优先级
# 能处理复杂规格如"500ml6瓶整箱"
# 更智能的无关信息过滤
# 自动识别和移除营销词、配送信息等
# 保留商品核心信息
# 更高的处理精度
# 多维度置信度评分
# 规则处理和大模型处理的优势结合
# 这套解决方案保持了原有代码的基本架构，同时通过引入NLP技术显著提升了数据清洗能力，尤其是在品牌识别、规格处理和无关信息过滤方面的表现大幅提升。

class Processor:
    """基础数据处理器，提供基本的文本清洗功能"""
    
    def __init__(self):
        self.punctuation = string.punctuation + '，。！？、；：""''（）【】《》…—「」『』〈〉'
    
    def process(self, text):
        """
        处理文本，执行基础清洗
        
        参数:
            text (str): 需要处理的文本
            
        返回:
            str: 处理后的文本
        """
        if not text or not isinstance(text, str):
            return ""
        
        # 标准化处理
        processed_text = self._normalize(text)
        
        # 移除多余空格
        processed_text = self._remove_extra_spaces(processed_text)
        
        # 移除特殊符号
        processed_text = self._remove_special_chars(processed_text)
        
        return processed_text
    
    def _normalize(self, text):
        """标准化文本，将全角字符转换为半角，统一大小写等"""
        # 将全角字符转换为半角
        text = unicodedata.normalize('NFKC', text)
        
        # 将英文字符转为小写
        text = re.sub(r'[A-Z]', lambda x: x.group(0).lower(), text)
        
        return text
    
    def _remove_extra_spaces(self, text):
        """移除多余的空格"""
        # 将多个空格替换为单个空格
        text = re.sub(r'\s+', ' ', text)
        
        # 移除首尾空格
        text = text.strip()
        
        return text
    
    def _remove_special_chars(self, text):
        """移除特殊字符"""
        # 将标点符号替换为空格
        for char in self.punctuation:
            text = text.replace(char, ' ')
        
        # 再次移除多余空格
        text = self._remove_extra_spaces(text)
        
        return text
    
    def tokenize(self, text):
        """
        简单的分词功能，仅使用空格分割
        
        参数:
            text (str): 需要分词的文本
            
        返回:
            list: 分词结果列表
        """
        # 预处理
        processed_text = self.process(text)
        
        # 使用空格进行简单分词
        tokens = processed_text.split()
        
        return tokens